package com.bird.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author sijunqiang
 * @data 2021/7/18 14:16
 * @description: kafka生产者发送消息
 */
public class ProducerKafka {

    private final static String TOPIC_NAME = "sijunqiang-topic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //生产者异步阻塞发给消息
        //producerAsyncSend();

        //异步发送消息且带回调函数
        //AsyncSendWithCallBack();

        //自定义partition 负载均衡
        asyncSendWithCallBackAndPartition();
    }

    /**
     * 生产者异步阻塞发给消息
     */
    public static void producerAsyncSend() throws ExecutionException, InterruptedException {

        // 属性配置对象
        Properties properties = new Properties();
        // kafka集群地址
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.223.128:9092,192.168.223.128:9093,192.168.223.128:9094");
        // 保证数据不会的丢失

        // 设置生产者发送消息的保证：

        // 这个配置可以保证消息确认机制。包括三种 0,1,all
        // 如果设置:acks=0  生产者将只发送一次消息到broker节点上,不管broker是否接收到消息都将不会再发送了，即使设置的重试的次数 也不过再次重发
        // 如果设置acks=1   生产者发送消息到主节点上，如果主节点收到消息并没有给到生产者回调的确认机制，那么生产者将会根据收设置重试的机制 再次发送。
        // 如果设置acks=all 当生产者发送消息到主节点上 那么就需要主节点和副本节点都完全同步了该消息 这个时候主节点才会给生产者一个确认机制
        // 这个是最强保证 也是性能最低的 因为broker 会做transactionalId的幂等性校验保证 消息只会有一次
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        // 设置生产者发送消息 如果失败了的重试次数。
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "0");
        // 设置批量发送的消息的大小
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "16384");

        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        // key+value序列化
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // producer的主对象
        Producer<String, String> producer = new KafkaProducer(properties);


        for (int i = 0; i < 10; i++) {
            String key = "key-" + i;
            // 构建消息对象+发送
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key-" + i, "value-" + i);
            Future<RecordMetadata> send = producer.send(record);

            // 异步阻塞发送消息 和join差不多 主线程需要等到子线程执行完毕之后 才能执行 而且还具有可见性。
            // 这里面是利用了多线程的阻塞方法 join
            RecordMetadata recordMetadata = send.get();
            System.out.println(key + "partition : " + recordMetadata.partition() + " , offset : " + recordMetadata.offset());
        }
        // 关闭所有的通道
        producer.close();
    }

    /**
     * 异步发送消息且带回调函数
     */
    public static void AsyncSendWithCallBack() throws ExecutionException, InterruptedException {

        // 属性配置对象
        Properties properties = new Properties();
        // kafka集群地址
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.223.128:9092,192.168.223.128:9093,192.168.223.128:9094");
        // 保证数据不会的丢失
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        // 设置生产者发送消息 如果失败了的重试次数。
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "0");
        // 设置批量发送的消息的大小
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "16384");

        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        // key+value序列化
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // producer的主对象
        Producer<String, String> producer = new KafkaProducer(properties);


        for (int i = 0; i < 10; i++) {
            String key = "key-" + i;
            // 构建消息对象+发送
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key-" + i, "value-" + i);
            Future<RecordMetadata> send = producer.send(record, new Callback() {
                // 当生产者发送完成 会执行这个回调函数 记录发送请求的日志
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    System.out.println(key + "partition : " + metadata.partition() + " , offset : " + metadata.offset());
                }
            });

        }
        // 关闭所有的通道
        producer.close();
    }


    /**
     * 异步发送消息且带回调函数 并且自定义partition负载均衡 将消息发送到不同的分区中
     */
    public static void asyncSendWithCallBackAndPartition() {


        // 属性配置对象
        Properties properties = new Properties();
        // kafka集群地址
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.223.128:9092,192.168.223.128:9093,192.168.223.128:9094");
        // 保证数据不会的丢失
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        // 设置生产者发送消息 如果失败了的重试次数。
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "0");
        // 设置批量发送的消息的大小
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "16384");

        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        // key+value序列化
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 加载自定义负载均衡 partition 不再使用默认的分区
        properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, " com.bird.producer.SimplePartition");

        // producer的主对象
        Producer<String, String> producer = new KafkaProducer(properties);


        for (int i = 0; i < 10; i++) {
            String key = "key-" + i;
            // 构建消息对象+发送
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key-" + i, "value-" + i);
            Future<RecordMetadata> send = producer.send(record, new Callback() {
                // 当生产者发送完成 会执行这个回调函数 记录发送请求的日志
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    System.out.println(key + "partition : " + metadata.partition() + " , offset : " + metadata.offset());
                }
            });

        }
        // 关闭所有的通道
        producer.close();
    }


}
